package transaction;

import cn.hutool.core.util.RandomUtil;

import cn.hutool.core.util.StrUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.nio.charset.StandardCharsets;

/**
 * 事务消息-生产者
 *
 * @author Administrator
 * @date 2022/04/01 14:53
 **/
@Slf4j
public class TransactionProducer {
    public static void main(String[] args) throws MQClientException {
        TransactionMQProducer producer = new TransactionMQProducer("Transaction-Producer");

        producer.setNamesrvAddr("rocketmq1.cn:9876;rocketmq2.cn:9876");


        /**
         * 提交半消息，mq回调去执行本地事务
         * 执行完本地事务，告知mq本事务的执行情况，是commit 还是rollback
         * 如果是commit 那么就会对消费者可见
         * 当一段时间mq没有收到消息的确认状态，那么mq就会回调check方法，去check本地事务
         * 会有最大的检查次数默认15次
         */
        producer.setTransactionListener(new TransactionListener() {

            /**
             * 执行本地事务
             * @param msg 消息
             * @param arg
             * @return
             */
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                if(StrUtil.equals(msg.getTags(),"Tag_A")){
                    // 提交
                    return LocalTransactionState.COMMIT_MESSAGE;
                }else if (StrUtil.equals(msg.getTags(), "Tag_B")) {
                    // 回滚
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }else {
                    // 不处理
                    return LocalTransactionState.UNKNOW;
                }

            }

            /**
             *
             * 超时之后，回调本地事务
             * @param msg
             * @return
             */
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                log.info("rocketmq 回调 tag:{}",msg.getTags());
//                if(StrUtil.equals(msg.getTags(),"Tag_C")){
//                    // 提交上去
//                    return LocalTransactionState.COMMIT_MESSAGE;
//                }
                return null;
            }
        });
        String[] tags = new String[]{"Tag_A","Tag_B","Tag_C"};

        producer.start();
        for (int i = 0; i < 3 ;i++){
            Message message = new Message("Transaction-Msg", tags[i], RandomUtil.randomString(100).getBytes(StandardCharsets.UTF_8));

            TransactionSendResult result = producer.sendMessageInTransaction(message, null);
            log.info("发送的结果:{}",result);
        }


    }


}
